package com.activemq;

import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.springframework.boot.autoconfigure.jms.JmsProperties;

import javax.jms.*;

/**
 * 消息订阅模式
 * 先启动订阅，再启动生产，不然消息就变成了废消息
 *
 * @author wangwy16
 * @version 1.0
 */
@Slf4j
public class MsgQueueSubscribe {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Topic topic = session.createTopic(MqHelloWorld.TOPIC);
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(message -> {
            if (message instanceof TextMessage) {
                try {
                    log.error("消息消费者接收消息：{}", ((TextMessage) message).getText());
                } catch (JMSException e) {
                    log.error("打印消息异常：{}", message, e);
                }
            }
        });
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
}

@Slf4j
class MsgQueueSubscribeB {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Topic topic = session.createTopic(MqHelloWorld.TOPIC);
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(message -> {
            if (message instanceof TextMessage) {
                try {
                    log.error("消息消费者B接收消息：{}", ((TextMessage) message).getText());
                } catch (JMSException e) {
                    log.error("消费者B打印消息异常：{}", message, e);
                }
            }
        });
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
}

@Slf4j
class MsgQueueSubscribeC {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Topic topic = session.createTopic(MqHelloWorld.TOPIC);
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(message -> {
            if (message instanceof TextMessage) {
                try {
                    log.error("消息消费者C接收消息：{}", ((TextMessage) message).getText());
                } catch (JMSException e) {
                    log.error("消费者C打印消息异常：{}", message, e);
                }
            }
        });
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
}

@Slf4j
class MsgSender {
    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(MqHelloWorld.URL);
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, JmsProperties.AcknowledgeMode.AUTO.getMode());
        Topic topic = session.createTopic(MqHelloWorld.TOPIC);
        MessageProducer producer = session.createProducer(topic);
        for (int i = 0; i < 6; i++) {
            TextMessage message = new ActiveMQTextMessage();
            message.setText("测试文本内容:" + i);
            producer.send(message);
        }
        producer.close();
        session.close();
        connection.close();
    }
}